package com.at.process13;

import com.at.bean.WaterSensor;
import com.at.functions5.WaterSensorMapFunction3;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

/**
 * @author huangchao E-mail:fengquan8866@163.com
 * @version 创建时间：2024/9/29 20:13
 */
public class KeyedProcessFunctionTopNDemo3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 7777)
                .map(new WaterSensorMapFunction3())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L)
                );

        /**
         * TODO 思路一：所有数据到一起，用hashmap存，key=vc，value= count值
         * 最近10秒 = 窗口长度，每5秒输出 = 滑动步长
         */
//        sensorDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
//                .process(
//                        new ProcessAllWindowTopNDemo2.MyTopNPAWF()
//                ).print();

        /**
         * TODO 思路二：使用 KeyedProcessFunction实现
         * 最近10秒 = 窗口长度，每5秒输出 = 滑动步长
         *
         * 1、按照vc做keyby，开窗，分别count
         *      =》 增量聚合，计算 count
         *      =》 全窗口，对计算结果 count值封装， 带上 窗口结束时间的 标签
         *          =》  为了让同一个窗口时间范围的计算结果到一起去
         *
         * 2、对同一个窗口范围的count值进行处理：排序、取前N个
         *      =》 按照 windowEnd做keyby
         *      =》 使用process， 来一条调用一次，需要先存，分开存 Hashmap，key=windowEnd，value=List
         *          =》 使用定时器，对 存起来的结果 进行 排序、取前N个
         */

        // 1.按照 vc 分组、开窗、聚合（增量计算+全量打标签）
        //   开窗聚合后，就是普通的流，没有了窗口信息，需要自己打上窗口的标记 windowEnd
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> windowAgg = sensorDS.keyBy(sensor -> sensor.getVc())
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(
                        new VcCountAgg(),
                        new WindowResult()
                );

        // 2、按照窗口标签（窗口结束时间）keyby，保证同一个窗口时间范围的结果，到一起。排序、取TopN
        windowAgg.keyBy(r -> r.f2)
                .process(new TopN(2))
                .print();

        env.execute();
    }

    public static class VcCountAgg implements AggregateFunction<WaterSensor, Integer, Integer> {

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(WaterSensor value, Integer accumulator) {
            return accumulator + 1;
        }

        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return null;
        }
    }

    /**
     * 泛型如下：
     * Param1：输入类型 = 增量函数的输出 count值，Integer
     * Param2：输出类型 = Tuple3（vc，count，windowEnd），带上 窗口结束时间 的标签
     * Param3：key类型，vc，Integer
     * Param4：窗口类型
     */
    public static class WindowResult extends ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow> {

        @Override
        public void process(Integer key, ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
            // 迭代器里面只有一条数据，next一次即可
            Integer count = elements.iterator().next();
            Long winEnd = context.window().getEnd();
            out.collect(Tuple3.of(key, count, winEnd));
        }
    }

    public static class TopN extends KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String> {

        // 存不同窗口的 统计结果，key=windowEnd，value=List数据
        private Map<Long, List<Tuple3<Integer, Integer, Long>>> dataListMap;

        // 要取的Top数量
        private int threshold;

        public TopN(int threshold) {
            this.threshold = threshold;
            dataListMap = new HashMap<>();
        }

        @Override
        public void processElement(Tuple3<Integer, Integer, Long> value,
                                   KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.Context ctx,
                                   Collector<String> out) throws Exception {
            // 进入这个方法，只有一条数据，要排序，得到齐才行 ===》 存起来，不同窗口分开存
            // 1、存到hashmap中
            Long windowEnd = value.f2;
            List<Tuple3<Integer, Integer, Long>> dataList = dataListMap.computeIfAbsent(windowEnd, k -> new ArrayList<>());
            dataList.add(value);

            // 2、注册一个定时器，windowEnd + 1ms即可
            //    同一个窗口范围，应该同时输出，只不过是一条一条调用processElement方法，只需要延迟10ms即可
            System.out.println("注册定时任务 vc=" + value.f0 + ", time=" + (windowEnd + 10));
            ctx.timerService().registerProcessingTimeTimer(windowEnd + 10);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            // 定时器触发，同一个窗口范围的计算结果攒齐了，开始 排序、取TopN
            Long windowEnd = ctx.getCurrentKey();
            // 1、 排序
            List<Tuple3<Integer, Integer, Long>> datas = dataListMap.get(windowEnd);
            datas.sort(new Comparator<Tuple3<Integer, Integer, Long>>() {
                @Override
                public int compare(Tuple3<Integer, Integer, Long> o1, Tuple3<Integer, Integer, Long> o2) {
                    // 降序，后 - 前
                    return o2.f1 - o1.f1;
                }
            });
            // 2、 取TopN
            StringBuilder outStr = new StringBuilder();
            outStr.append("==========================");
            // 遍历 排序后的list，取出前 threshold 个， 考虑可能list不够2个的情况 ==》 list中元素的个数 和2 取最小值
            for (int i = 0; i < Math.min(threshold, datas.size()); i++) {
                Tuple3<Integer, Integer, Long> vcCount = datas.get(i);
                outStr.append("\n");
                outStr.append("Top").append(i + 1).append("的vc");
                outStr.append("\n");
                outStr.append("vc=").append(vcCount.f0);
                outStr.append("\n");
                outStr.append("count=").append(vcCount.f1);
                outStr.append("\n");
                outStr.append("窗口结束时间=").append(vcCount.f2);
                outStr.append("\n");
            }
            outStr.append("=======================================");

            // 用完的list，及时清理，节省资源
            dataListMap.clear();

            out.collect(outStr.toString());
        }
    }
}
